package zk_01.zookeeper.sample;

import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class WorkerUnSyn implements Watcher{
	
	private static Logger logger = Logger.getLogger(WorkerUnSyn.class);
	
	private String hostPort;
	private ZooKeeper zk;
	private String status;
	private String serverId;
	private String name;
	private ExecutorService es ;

	public static void main(String[] args) {
		//WorkerUnSyn worker = new WorkerUnSyn("10.40.12.196:2181");
		WorkerUnSyn worker = new WorkerUnSyn("192.168.1.113:2181");
		try {
			worker.startZK();
			worker.register();
			Thread.sleep(1500000);
			worker.stopZK();
			worker.closePool();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	public void closePool(){
		es.shutdown();
	}
	
	StatCallback statusUpdateCallback = new StatCallback(){

		public void processResult(int rc, String path, Object ctx, Stat stat) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				updateStatus(new String((byte[])ctx));
				break;
			}
		}
	};
	
	StringCallback createWorkerCallback = new StringCallback(){

		public void processResult(int rc, String path, Object ctx, String name) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				register();
				break;
			case OK:
				logger.info("registered successfully:"+name);
				addAssignWorker((String)ctx);
				break;
			case NODEEXISTS:
				logger.info("already registered:"+name);
				break;
			default:
				logger.info("something was wrong:"+KeeperException.create(Code.get(rc),path));	
			}
		}
		
	};
	
	StringCallback addAssignWorkerCallback = new StringCallback(){

		public void processResult(int rc, String path, Object ctx, String name) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				addAssignWorker((String)ctx);
				break;
			case OK:
				logger.info("add worker "+(String)ctx+" to assign znode successfully:");
				getTaskByName((String)ctx);
				break;
			case NODEEXISTS:
				logger.info("worker "+(String)ctx+" already exists");
				break;
			default:
				logger.info("something was wrong:"+KeeperException.create(Code.get(rc),path));	
			}
		}
	};
	
	public void addAssignWorker(String workerName){
		zk.create("/assign/"+workerName, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, addAssignWorkerCallback, workerName);
	}
	
	public WorkerUnSyn() {
	}
	
	public WorkerUnSyn(String hostPort){
		this.hostPort=hostPort;
		this.es = Executors.newFixedThreadPool(10);
	}
	
	public void startZK() throws IOException{
		zk = new ZooKeeper(hostPort,15000,this);
	}
	
	public void stopZK() throws InterruptedException{
		zk.close();
	}
	
	public void register(){
		Random random = new Random();
		serverId = ""+random.nextInt(10);
		name = "worker-"+serverId;
		logger.info("workerName is:"+name);
		zk.create("/workers/worker-"+serverId, "Idle".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, createWorkerCallback, name);
	}

	public void process(WatchedEvent event) {
		logger.info(event);
	}
	
	public void setStatus(String status){
		this.status=status;
		updateStatus(status);
	}
	
	public void updateStatus(String status){
		if(status==this.status){
			zk.setData("/workers/"+name, status.getBytes(), -1,statusUpdateCallback,status);
		}
	}
	
	private void getTaskByName(String sName){
		List<String> lTask = null;
		zk.getChildren("/assign/"+sName, getTaskWatcher, getDistributedTaskCallback, sName);
	}
	
	Watcher getTaskWatcher = new Watcher(){

		public void process(WatchedEvent event) {
			if(event.getType()==EventType.NodeChildrenChanged){
				getTaskByName(name);
			}
		}
		
	};
	
	ChildrenCallback getDistributedTaskCallback = new ChildrenCallback(){

		public void processResult(int rc, String path, Object ctx,
				List<String> children) {
			//logger.info("getDistributedTaskCallback path:"+path);
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getTaskByName((String)ctx);
				break;
			case OK:
				distributeAssignedTask(children,(String)ctx);
				break;
			default:
				logger.info("getDistributedTaskCallback failed , "+Code.get(rc));
			}
		}
		
	};
	
	public void distributeAssignedTask(List<String> children,String workerPath){
		//分配任务给从节点
		for(String tName:children){
			getAssignedTaskData(workerPath+"/"+tName,tName);
		}
		
		//待已失效的从节点对应任务重新分配完毕，从/assign/下删除对应从节点
		//##########################################待补充###################################################
	}

	public void getAssignedTaskData(String tPath,String tName){
		zk.getData("/assign/"+tPath, false, getAssignedTaskDataCallback, tName);
	}
	
	DataCallback getAssignedTaskDataCallback = new DataCallback(){

		public void processResult(int rc, String path, Object ctx, byte[] data,
				Stat stat) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getAssignedTaskData(path,(String)ctx);
				break;
			case OK:
				runTask(path,new String(data),stat.getVersion());
				break;
			default:
				logger.info("getAssignedTaskDataCallback failed");
				break;
			}
		}
	};
	
	public void runTask(final String path,final String tContent,final int version){
		es.submit(new Runnable() {
			public void run() {
				logger.info("Task "+tContent+"'s running was end");
				deleteTask(path,version);
			}
		});
	}
	public void deleteTask(String path,int version){
		zk.delete(path, version, deleteTaskCallback, new Integer(version));
	}
	VoidCallback deleteTaskCallback = new VoidCallback(){

		public void processResult(int rc, String path, Object ctx) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				deleteTask(path,((Integer)ctx).intValue());
				break;
			case OK:
				logger.info(path+" znode was deleted from zk");
				break;
			default:
				logger.info("deleteTaskCallback failed");
				break;
			}
		}
		
	};
}
